package org.ala.tiktools.rpc.client_server.client;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rpc.config.RpcConfigInfo;
import org.ala.tiktools.rpc.exception.RpcNoLoginException;
import org.ala.tiktools.rpc.exception.RpcRemoteException;
import org.ala.tiktools.rpc.exception.RpcRuntimeException;
import org.ala.tiktools.rpc.msg.ack.RpcAckBody;
import org.ala.tiktools.rpc.plugin.circuit_breaker.CircuitBreaker;
import org.ala.tiktools.rpc.plugin.circuit_breaker.ICircuitBreaker;
import org.ala.tiktools.rpc.plugin.interceptor.IRpcInterceptorChain;
import org.ala.tiktools.rpc.plugin.interceptor.RpcInterceptor;
import org.ala.tiktools.rpc.plugin.load_balance.ILoadBalance;
import org.ala.tiktools.rpc.plugin.node.INode;
import org.ala.tiktools.rpc.plugin.rate_limiter.IRateLimiter;
import org.ala.tiktools.rpc.plugin.rate_limiter.RateLimiter;
import org.ala.tiktools.rpc.spring.RpcBeanConfiguration;
import org.ala.tiktools.tools.BeanTools;
import org.ala.tiktools.tools.CachingKVTools;
import org.slf4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;


/**
 * rpc 客户端代理
 * <p>  一个RpcClient对应一个实例
 * <p>  持有loadBalance，通过loadBalance拿通讯器，发消息
 * <p>  执行限流器逻辑
 * <p>  执行断路器逻辑
 *
 * @author ala
 * @date 2025-04-04 14:24
 */
public class RpcClientProxy implements InvocationHandler {
	static Logger log = LoggerFactory.getLogger("rpc", "client");
	
	
    /**
     *  节点组
     *  <p> 优先取注解的name，其次取hosts
     */
    protected String group;
    /**
     *  根url
     */
    protected String rootUrl;
    /**
     *  spring环境
     */
    protected ApplicationContext context;
    /**
     *  loadBalance 负载均衡器
     *  <p> 名字叫的高大上，其实就一个select。。。
     */
    protected ILoadBalance loadBalance;
    /**
     * 	与方法绑定的拦截器链
     */
    protected CachingKVTools<Method, IRpcInterceptorChain> chainMapper = new CachingKVTools<Method, IRpcInterceptorChain>(IRpcInterceptorChain.NULL, m -> {
    	RpcInterceptor anno = m.getAnnotation(RpcInterceptor.class);
    	IRpcInterceptorChain chain = null;
    	//	优先取自己配的拦截器
    	if (anno != null && anno.chain() != null) {
    		chain = BeanTools.getBean(context, anno.chain());
    	}
    	//	取不到取默认拦截器
    	if (chain == null) {
    		chain = BeanTools.getBean(context, RpcBeanConfiguration.BEAN_INTERCEPTOR_SENDER_CHAIN, IRpcInterceptorChain.class);
    	}
    	return chain;
    });
    
    
    /**
     *  与方法绑定的url
     */
    protected CachingKVTools<Method, String> methodUrlMapper = new CachingKVTools<>("NULL", m -> {
        PostMapping postAnno = m.getAnnotation(PostMapping.class);
        String url = null;
        if (postAnno != null && postAnno.value().length > 0) {
            url = postAnno.value()[0];
        }
        GetMapping getAnno = m.getAnnotation(GetMapping.class);
        if (getAnno != null && getAnno.value().length > 0) {
            url = getAnno.value()[0];
        }
        PutMapping putAnno = m.getAnnotation(PutMapping.class);
        if (putAnno != null && putAnno.value().length > 0) {
            url = putAnno.value()[0];
        }
        if (StringUtils.isEmpty(url)) {return null;}
        if (url.charAt(0) == '/') {
            url = url.substring(1);
        }
        return url;
    });
    /**
     *  方法对应的断路器
     */
    protected CachingKVTools<Method, ICircuitBreaker> methodCircuitBreakerMapper = new CachingKVTools<>(ICircuitBreaker.NULL, m -> {
        CircuitBreaker anno = m.getAnnotation(CircuitBreaker.class);
        if (anno == null) {return null;}
        ICircuitBreaker cb = BeanTools.getBeanOrNew(context, anno.clazz());
        cb.setApplicationContext(context);
        return cb;
    });
    /**
     *  方法对应的限流器
     */
    protected CachingKVTools<Method, IRateLimiter> methodRateLimiter = new CachingKVTools<>(IRateLimiter.NULL, m -> {
        RateLimiter anno = m.getAnnotation(RateLimiter.class);
        if (anno == null) {return null;}
        IRateLimiter rl = BeanTools.getBeanOrNew(context, anno.clazz());
        rl.setApplicationContext(context);
        return rl;
    });


    /**
     *  执行请求
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        //  从方法体上拿url
        String url = methodUrlMapper.get(method);
        if (StringUtils.isEmpty(url)) {
            return method.invoke(this, args);
        }
        url = rootUrl + "/" + url;
        
        //	尝试执行方法拦截器
        IRpcInterceptorChain chain = chainMapper.get(method);
        Map<String, Object> metadata = null;
        if (chain != null) {
        	metadata = new HashMap<>();
        	chain.doInterceptor(method, args, metadata);
        }

        //  断路器
        ICircuitBreaker circuitBreaker = methodCircuitBreakerMapper.get(method);
        //  限流器
        IRateLimiter rateLimiter = methodRateLimiter.get(method);
        
        //  如果是异步执行
        if (Future.class.isAssignableFrom(method.getReturnType())) {
        	CompletableFuture<Object> cf = new CompletableFuture<>();
        	sendAsync(method, circuitBreaker, rateLimiter, url, args, metadata, (ack, e) -> {
        		if (ack != null) {
        			if (ack.isSuccess()) {
        				//	尝试执行后置拦截
        				Object ret = ack.getReturnValues();
        				if (chain != null) { 
        					try {
        						chain.afterInterceptor(method, args, ret, ack.getMetadata()); 
        					} catch (Exception e1) {
        						//	后置拦截过程中抛出任何异常均不予处理
        						log.error("[RpcClientProxy invoke] 后置拦截出现异常，但不打断原方法执行", e1);
        					}
        				}
        				
        				cf.complete(ret);
        			} else {
        				cf.completeExceptionally(new RpcRemoteException(ack.getErrorMsg()));
        			}
        		} else {
        			cf.completeExceptionally(e != null ? e : new RpcRemoteException());
        		}
        	});
        	return cf;
        }
        //	如果是同步执行
        else {
        	RpcAckBody ack = send(method, circuitBreaker, rateLimiter, url, args, metadata);
        	if (ack.isSuccess()) {
        		//	尝试执行后置拦截
        		Object ret = ack.getReturnValues();
				if (chain != null) { 
					try {
						chain.afterInterceptor(method, args, ret, ack.getMetadata()); 
					} catch (Exception e1) {
						//	后置拦截过程中抛出任何异常均不予处理
						log.error("[RpcClientProxy invoke] 后置拦截出现异常，但不打断原方法执行", e1);
					}
				}
				
        		return ret;
        	} else {
        		throw new RpcRemoteException(ack.getErrorMsg());
        	}
        }
    }
    /**
     * 	取链接，同步执行请求
     */
    protected RpcAckBody send(Method method, ICircuitBreaker circuitBreaker, IRateLimiter rateLimiter, String url, Object[] args, Map<String, Object> metadata) throws Throwable {
        //  执行远程
        INode node = loadBalance.select(group);
        //  如果此时取不到节点，执行断路器
        if (node == null) {
            if (circuitBreaker != null) {
                circuitBreaker.fallbackNoActiveNode(method, args);
            } else {
                throw new RpcRuntimeException(String.format("当前没有有效节点 group:%s", group));
            }
            return null;
        }
        
        //	执行断路器
        //  如果此时断路器返回true，直接执行断路器
        if (circuitBreaker != null && circuitBreaker.circuit(method, args, node.key())) {
            circuitBreaker.fallback(method, args, node.key());
            return null;
        }
        
        //	执行限流
        if (rateLimiter != null) {
            long limit = rateLimiter.limit(args, node.key());
            //  10ms以内直接自旋，否则sleep
            if (limit <= 10) {
                long start = System.currentTimeMillis();
                while (System.currentTimeMillis() - start < limit) {}
            } else {
                Thread.sleep(limit);
            }
        }
        
        //	同步执行
        try {
            //  如果method没有返回值，直接异步执行掉
            if (Void.TYPE == method.getReturnType()) {
                node.sendVoid(url, args);
                return null;
            } 
        	CompletableFuture<RpcAckBody> cf = node.send(url, args, metadata);
        	return cf.get(RpcConfigInfo.INSTANCE.getTimeout(), TimeUnit.MILLISECONDS);
        } catch (Throwable e) {
        	if (e instanceof ExecutionException) {
        		Throwable ex = e.getCause();
        		//	如果连接还未登录（udp下可能出现，因为无法感知对方链接状态，可能重启了登录状态都清掉了，尝试重连后继续执行）
        		if (ex instanceof RpcNoLoginException) {
        			log.warn("[RpcClientProxy invoke] 链接登录失效，先用其他链接执行，尝试重新登录 node:" + node.key());
        			//	active会关闭连接，关闭后会有监听从活跃列表中清除
        			node.active();
        			return send(method, circuitBreaker, rateLimiter, url, args, metadata);
            	}
        	}
        	throw e;
        }
    }
    /**
     * 	取链接，异步执行
     */
    protected void sendAsync(Method method, ICircuitBreaker circuitBreaker, IRateLimiter rateLimiter, String url, Object[] args, Map<String, Object> metadata, BiConsumer<RpcAckBody, Throwable> consumer) throws Throwable {
        //  执行远程
        INode node = loadBalance.select(group);
        //  如果此时取不到节点，执行断路器
        if (node == null) {
            if (circuitBreaker != null) {
                circuitBreaker.fallbackNoActiveNode(method, args);
            } else {
            	consumer.accept(null, new RpcRuntimeException(String.format("当前没有有效节点 group:%s", group)));
            }
        }
        
        //	执行断路器
        //  如果此时断路器返回true，直接执行断路器
        if (circuitBreaker != null && circuitBreaker.circuit(method, args, node.key())) {
            circuitBreaker.fallback(method, args, node.key());
            //	当执行逻辑断路时，没有异常抛出。所以代码中要兼容这种情况
            consumer.accept(null, null);
            return;
        }
        
        //	执行限流
        if (rateLimiter != null) {
            long limit = rateLimiter.limit(args, node.key());
            //  10ms以内直接自旋，否则sleep
            if (limit <= 10) {
                long start = System.currentTimeMillis();
                while (System.currentTimeMillis() - start < limit) {}
            } else {
                Thread.sleep(limit);
            }
        }
        
        try {
            //  如果method没有返回值，直接异步执行掉
            if (Void.TYPE == method.getReturnType()) {
                node.sendVoid(url, args);
                return;
            } 
        	node.send(url, args, metadata).whenCompleteAsync(consumer);
        } catch (Throwable e) {
        	if (e instanceof ExecutionException) {
        		Throwable ex = e.getCause();
        		//	如果连接还未登录（udp下可能出现，因为无法感知对方链接状态，可能重启了登录状态都清掉了，尝试重连后继续执行）
        		if (ex instanceof RpcNoLoginException) {
        			log.warn("[RpcClientProxy invoke] 链接登录失效，先用其他链接执行，尝试重新登录 node:" + node.key());
        			//	active会关闭连接，关闭后会有监听从活跃列表中清除
        			node.active();
        			sendAsync(method, circuitBreaker, rateLimiter, url, args, metadata, consumer);
            	}
        	}
        	consumer.accept(null, e);
        }
//        //	如果method的返回值是Future类型，执行异步（只认这个，别的不认）
//        else if (CompletableFuture.class.equals(method.getReturnType())) {
//        	
//        }
    }
    

    public String toString() {
        return String.format("RpcClientProxy{group:%s}", group);
    }


    public ILoadBalance getLoadBalance() {
        return loadBalance;
    }

    public void setLoadBalance(ILoadBalance loadBalance) {
        this.loadBalance = loadBalance;
    }

    public ApplicationContext getContext() {
        return context;
    }

    public void setContext(ApplicationContext context) {
        this.context = context;
    }

    public String getGroup() {
        return group;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public String getRootUrl() {
        return rootUrl;
    }

    public void setRootUrl(String rootUrl) {
        if (!StringUtils.isEmpty(rootUrl) && rootUrl.charAt(rootUrl.length()-1) == '/') {
            rootUrl = rootUrl.substring(0, rootUrl.length()-1);
        }
        this.rootUrl = rootUrl;
    }
}
